package org.apache.ocean;

import java.io.InputStream;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.logging.Logger;

import org.apache.commons.lang.StringUtils;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.IndexOutput;
import org.apache.ocean.snapshot.Snapshot;
import org.apache.ocean.snapshot.Snapshot.SnapshotFile;

public class SnapshotReplicationClient {
	public static Logger log = Logger.getLogger(SnapshotReplicationClient.class.getName());
	ClusterIndexModifier clusterIndexModifier;
	Snapshot snapshot;
	URL url;

	public SnapshotReplicationClient(URL url, Snapshot snapshot, ClusterIndexModifier clusterIndexModifier) throws Exception {
		this.url = url;
		this.snapshot = snapshot;
		this.clusterIndexModifier = clusterIndexModifier;
		run();
		clusterIndexModifier.reload();
	}

	private void run() throws Exception {
		Directory directory = clusterIndexModifier.getDirectory();
		List<String> localToDelete = new ArrayList<String>();
		List<String> localToAdd = new ArrayList<String>();
		Set<String> localFileNames = toStringSet(directory.list());
		for (String localFileName : localFileNames) {
			if (!snapshot.files.containsKey(localFileName)) {
				localToDelete.add(localFileName);
			}
		}
		for (String remoteFileName : snapshot.files.keySet()) {
			if (!localFileNames.contains(remoteFileName)) {
				if (!StringUtils.equals(remoteFileName, snapshot.name)) {
					localToAdd.add(remoteFileName);
				}
			}
		}
		downloadFiles(localToAdd);
	}

	private void downloadFiles(List<String> fileNames) throws Exception {
		int errorCount = 0;
		for (String fileName : fileNames) {
			if (errorCount > 7) {
				throw new Exception("errorCount: " + errorCount);
			}
			for (int x = 0; x < 2; x++) {
				try {
					SnapshotFile snapshotFile = snapshot.files.get(fileName);
					downloadFile(snapshotFile.name, snapshotFile.size.value);
				} catch (Exception exception) {
					log.severe(exception.getMessage());
					errorCount++;
				}
			}
		}
	}

	private void downloadFile(String fileName, long fileLength) throws Exception {
		Directory directory = clusterIndexModifier.getDirectory();
		HttpRequest httpRequest = new HttpRequest();
		CHttpClient.StreamResponse streamResponse = CHttpClient.getStreamResponse(httpRequest);
		try {
			IndexOutput indexOutput = clusterIndexModifier.getIndexFileOutput(fileName);
			InputStream input = streamResponse.getInputStream();
			byte[] buffer = new byte[16384];
			int length = 0;
			while ((length = input.read(buffer)) > 0) {
				indexOutput.writeBytes(buffer, length);
			}
			indexOutput.close();
			streamResponse.close();
			long directoryFileLength = directory.fileLength(fileName);
			if (directoryFileLength != fileLength) {
				directory.deleteFile(fileName);
				throw new Exception("directoryFileLength: " + directoryFileLength + " fileLength: " + fileLength);
			}
		} finally {
			if (streamResponse != null)
				streamResponse.close();
		}
	}

	private Set<String> toStringSet(String[] array) {
		Set<String> set = new HashSet<String>(array.length);
		for (String string : array) {
			set.add(string);
		}
		return set;
	}
}
